package com.ruins.ping.controller;

import cn.hutool.core.date.DateUtil;
import com.ruins.ping.producer.LogSender;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.reactive.function.client.WebClient;

import javax.annotation.Resource;
import java.util.Date;
import java.util.concurrent.TimeUnit;

@Slf4j
@RestController
@RequestMapping("ping")
public class PingController {

    private static final String PING_REQUEST_KEY_PREFIX = "ping_request_";

    private static final String PING_DISTRIBUTED_LOCK_KEY = "ping_distributed_lock";

    @Resource
    private RedisTemplate<String, Long> redisTemplate;

    @Resource
    private RedissonClient redissonClient;

    @Resource
    private WebClient webClient;

    @Resource
    private LogSender logSender;

    @Value("${scheduler.enabled:true}")
    private boolean isScheduledEnabled;

    @RequestMapping("/sendPing")
    public String sendPing() throws InterruptedException {
        if (!getPermission()) {
            String message = "ping limit";
            log.warn(message);
            logSender.sendLog(message);
            return "ping limit";
        }

        webClient.get().uri("/pong").retrieve()
                .bodyToMono(String.class)
                .doOnError(e -> {
                    String errorMessage = "ping false: " + e.getMessage();
                    log.error(errorMessage);
                    logSender.sendLog(errorMessage);
                })
                .doOnNext(response -> {
                    String infoMessage = "ping success: " + response;
                    log.info(infoMessage);
                    logSender.sendLog(infoMessage);
                })
                .subscribe();

        return "already send";
    }

    public boolean getPermission() throws InterruptedException {
        RLock lock = redissonClient.getLock(PING_DISTRIBUTED_LOCK_KEY);
        if (lock.tryLock(500, 500, TimeUnit.MILLISECONDS)) {
            try {
                Date unitDate = new Date();
                String dateFormat = DateUtil.format(unitDate, "yyyyMMddHHmmss");
                double source = Double.parseDouble(dateFormat);
                long currentTime = new Date().getTime();

                redisTemplate.opsForZSet().add(PING_REQUEST_KEY_PREFIX, currentTime, source);

                Long count = redisTemplate.opsForZSet().count(PING_REQUEST_KEY_PREFIX, source, source);

                this.cleanDataBeforeFiveSeconds(unitDate);

                return count == null || count <= 2;
            } finally {
                lock.unlock();
            }
        } else {
            return false;
        }
    }

    private void cleanDataBeforeFiveSeconds(Date unitDate) {
        long currentTimeMillis = unitDate.getTime();
        long fiveSecondsAgoMillis = currentTimeMillis - 5000;
        Date cleanDate = new Date(fiveSecondsAgoMillis);
        String dateFormat = DateUtil.format(cleanDate, "yyyyMMddHHmmss");
        double source = Double.parseDouble(dateFormat);

        redisTemplate.opsForZSet().removeRangeByScore(PING_REQUEST_KEY_PREFIX, 0, source);
    }

    @Scheduled(cron = "0/1 * * * * *")
    public void pingTest() throws InterruptedException {
        if (isScheduledEnabled) {
            this.sendPing();
        }
//        boolean b = getPermission();
//        if (b) {
//            log.info("true");
//        } else {
//            log.error("false");
//        }
    }
}